package com.health.passbook.service.impl;

import com.alibaba.fastjson.JSON;
import com.health.passbook.constant.Constants;
import com.health.passbook.mapper.PassTemplateRowMapper;
import com.health.passbook.service.IGainPassTemplateService;
import com.health.passbook.utils.RowKeyGenUtil;
import com.health.passbook.vo.GainPassTemplateRequest;
import com.health.passbook.vo.PassTemplate;
import com.health.passbook.vo.Response;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.avro.data.Json;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.hibernate.transform.PassThroughResultTransformer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * <h1>用户领取优惠券功能实现</h1>
 */
@Slf4j
@Service
public class GainPassTemplateServiceImpl implements IGainPassTemplateService {

    /** HBase 客户端 */
    private final HbaseTemplate hbaseTemplate;

    /** redis 客户端 */
    private final StringRedisTemplate redisTemplate;

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @Autowired
    public GainPassTemplateServiceImpl(HbaseTemplate hbaseTemplate,
                                       StringRedisTemplate redisTemplate) {
        this.hbaseTemplate = hbaseTemplate;
        this.redisTemplate = redisTemplate;
    }

    @Override
    public Response gainPassTemplate(GainPassTemplateRequest request) throws Exception {
        //获取redis分布式锁  setIfAbsent如果键不存在则新增,存在则不改变已经有的值。
//        Boolean  flag = redisTemplate.opsForValue().setIfAbsent(Constants.REDIS_LOCK, "1", 30,
//                TimeUnit.SECONDS);
        //释放锁可以放在try finally块中实现
        //如果第一次获取失败，即flag=false,那么会允许当前请求请求3次，如果3次没有工程，直接返回
//        int count=0;
//        while (!flag) {
//            if(count==2){
//                return new Response().failure("获取锁失败");
//            }
//            Thread.currentThread().sleep(1000);
//            flag = redisTemplate.opsForValue().setIfAbsent(Constants.REDIS_LOCK, "1",
//                    30, TimeUnit.SECONDS);
//            count++;
//        }
        PassTemplate passTemplate;
        String passTemplateId = RowKeyGenUtil.genPassTemplateRowKey(
                request.getPassTemplate());
        //先从redis中获取对象进行遇见优惠券库存操作
//        String json = redisTemplate.opsForValue().get(passTemplateId);
//        PassTemplate passTemplate1 = JSON.parseObject(json, PassTemplate.class);
//        if(passTemplate1.getLimit()<1){
//            Response response = new Response();
//            response.setData("优惠券已经领完");
//        }
//        passTemplate1.setLimit(passTemplate1.getLimit()-1);
        //存储到redis中实现遇见库存操作
        // json=JSON.toJSONString(passTemplate1);
        // redisTemplate.opsForValue().set(passTemplateId,json);
        //this.kafkaTemplate.send("REDUCE_STORE_COUPON",passTemplateId);

        //扣减库存的服务监听kafka中REDUCE_STORE_COUPON，拿到rowkey,进行库存更新操作


        try {
            passTemplate = hbaseTemplate.get(
                    Constants.PassTemplateTable.TABLE_NAME,
                    passTemplateId,
                    new PassTemplateRowMapper()
            );
        } catch (Exception ex) {
            log.error("Gain PassTemplate Error: {}",
                    JSON.toJSONString(request.getPassTemplate()));
            //释放锁
            //this.redisTemplate.delete(Constants.REDIS_LOCK);
            return Response.failure("Gain PassTemplate Error!");
        }

        if (passTemplate.getLimit() <= 1 && passTemplate.getLimit() != -1) {
            log.error("PassTemplate Limit Max: {}",
                    JSON.toJSONString(request.getPassTemplate()));
            //释放锁
            //this.redisTemplate.delete(Constants.REDIS_LOCK);
            return Response.failure("PassTemplate Limit Max!");
        }

        Date cur = new Date();
        if (!(cur.getTime() >= passTemplate.getStart().getTime()
                && cur.getTime() < passTemplate.getEnd().getTime())) {
            log.error("PassTemplate ValidTime Error: {}",
                    JSON.toJSONString(request.getPassTemplate()));
            //释放锁
            //this.redisTemplate.delete(Constants.REDIS_LOCK);
            return Response.failure("PassTemplate ValidTime Error!");
        }

        // 减去优惠券的 limit
        if (passTemplate.getLimit() != -1) {
            List<Mutation> datas = new ArrayList<>();
            byte[] FAMILY_C = Constants.PassTemplateTable.FAMILY_C.getBytes();
            byte[] LIMIT = Constants.PassTemplateTable.LIMIT.getBytes();
            Put put = new Put(Bytes.toBytes(passTemplateId));
            put.addColumn(FAMILY_C, LIMIT,
                    Bytes.toBytes(passTemplate.getLimit() - 1));
            datas.add(put);
            hbaseTemplate.saveOrUpdates(Constants.PassTemplateTable.TABLE_NAME,
                    datas);
        }

        // 将优惠券保存到用户优惠券表
        if (!addPassForUser(request, passTemplate.getId(), passTemplateId)) {
            //释放锁
           // this.redisTemplate.delete(Constants.REDIS_LOCK);
            return Response.failure("GainPassTemplate Failure!");
        }
        //释放锁
        // this.redisTemplate.delete(Constants.REDIS_LOCK);
        return Response.success();
    }

    @Override
    public Response updatePassTemlate(PassTemplate passTemplate, String rowkey) {

        //仅更新有效期，其他字段略
        byte[] FAMILY_C = Constants.PassTemplateTable.FAMILY_C.getBytes();
        byte[] START = Constants.PassTemplateTable.START.getBytes();
        byte[] END = Constants.PassTemplateTable.END.getBytes();

        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(
                Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constants.PassTemplateTable.START),
                Bytes.toBytes(org.apache.commons.lang3.time.DateFormatUtils.ISO_DATE_FORMAT.format(passTemplate.getStart()))
        );
        put.addColumn(
                Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constants.PassTemplateTable.END),
                Bytes.toBytes(org.apache.commons.lang3.time.DateFormatUtils.ISO_DATE_FORMAT.format(passTemplate.getEnd()))
        );
        try {
            this.hbaseTemplate.saveOrUpdate(Constants.PassTemplateTable.TABLE_NAME,put);
            return new Response().success();
        } catch (Exception e) {
            e.printStackTrace();
            return new Response().failure("更新失败");
        }
    }

    /**
     * <h2>给用户添加优惠券</h2>
     * @param request {@link GainPassTemplateRequest}
     * @param merchantsId 商户 id
     * @param passTemplateId 优惠券 id
     * @return true/false
     * */
    private boolean addPassForUser(GainPassTemplateRequest request,
                                   Integer merchantsId, String passTemplateId) throws Exception {

        byte[] FAMILY_I = Constants.PassTable.FAMILY_I.getBytes();
        byte[] USER_ID = Constants.PassTable.USER_ID.getBytes();
        byte[] TEMPLATE_ID = Constants.PassTable.TEMPLATE_ID.getBytes();
        byte[] TOKEN = Constants.PassTable.TOKEN.getBytes();
        byte[] ASSIGNED_DATE = Constants.PassTable.ASSIGNED_DATE.getBytes();
        byte[] CON_DATE = Constants.PassTable.CON_DATE.getBytes();

        List<Mutation> datas = new ArrayList<>();
        Put put = new Put(Bytes.toBytes(RowKeyGenUtil.genPassRowKey(request)));
        put.addColumn(FAMILY_I, USER_ID, Bytes.toBytes(request.getUserId()));
        put.addColumn(FAMILY_I, TEMPLATE_ID, Bytes.toBytes(passTemplateId));

        if (request.getPassTemplate().getHasToken()) {
            String token = redisTemplate.opsForSet().pop(passTemplateId);
            if (null == token) {
                log.error("Token not exist: {}", passTemplateId);
                return false;
            }
            recordTokenToFile(merchantsId, passTemplateId, token);
            put.addColumn(FAMILY_I, TOKEN, Bytes.toBytes(token));
        } else {
            put.addColumn(FAMILY_I, TOKEN, Bytes.toBytes("-1"));
        }

        put.addColumn(FAMILY_I, ASSIGNED_DATE,
                Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(new Date())));
        put.addColumn(FAMILY_I, CON_DATE, Bytes.toBytes("-1"));

        datas.add(put);

        hbaseTemplate.saveOrUpdates(Constants.PassTable.TABLE_NAME, datas);

        return true;
    }

    /**
     * <h2>将已使用的 token 记录到文件中</h2>
     * @param merchantsId 商户 id
     * @param passTemplateId 优惠券 id
     * @param token 分配的优惠券 token
     * */
    private void recordTokenToFile(Integer merchantsId, String passTemplateId,
                                   String token) throws Exception {

        Files.write(
                Paths.get(Constants.TOKEN_DIR, String.valueOf(merchantsId),
                        passTemplateId + Constants.USED_TOKEN_SUFFIX),
                (token + "\n").getBytes(),
                StandardOpenOption.CREATE, StandardOpenOption.APPEND
        );
    }
}
